package groupOne.app.DWD.db;

import groupOne.app.BaseAppSql;
import groupOne.common.Constant;
import groupOne.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


public class DwdTradeCartAdd_lyy extends BaseAppSql {
    public static void main(String[] args) { //加购事务事实表，在业务数据表dos_db中，从ods拿数据，
        new DwdTradeCartAdd_lyy().init(3001, 2, "DwdTradeCartAdd");

    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {



        //1、读取读取ods_db数据
        readOdsDb(tEnv,"DwdTradeCartAdd");

        //2、过滤cart_info数据
        Table result = tEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['course_id'] course_id, " +
                "data['course_name'] course_name, " +
                "data['cart_price'] cart_price, " +
                "data['session_id'] session_id, " +
                "data['create_time'] create_time, " +
                "data['update_time'] update_time, " +
                "data['deleted'] deleted, " +
                "data['sold'] sold, " +
                "ts " + //时间戳
                //  "pt " + //处理时间
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='cart_info' " +
                "and (" +
                " `type`='insert' " +
                "  or (`type`='update' " +
                "      and `old`['update_time'] is null " + // 由于sku_num变化导致的update
                "      and `old`['sold'] ='0' " +
                "     )" +
                ")");

        // tEnv.toChangelogStream(result).print();  //查看是否有数据，打印在控制台



        // 3、定义一个动态表与Kafka的topic相关联
        tEnv.executeSql("create table dwd_trade_cart_add( " +
                "id string, " +
                "user_id string, " +
                "course_id string, " +
                "course_name string, " +
                "cart_price string, " +
                "session_id string, " +
                "create_time string, " +
                "update_time string, " +
                "deleted string, " +
                "sold string, " +
                "ts bigint "+
                ")"+ SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_CART_ADD)
        );


        result.executeInsert("dwd_trade_cart_add");


    };


    //读取ods_db数据
    public void readOdsDb(StreamTableEnvironment stream,String groupId){
        stream.executeSql("create table ods_db(" +
                " `database` string, " +
                " `table` string, " +
                " `type` string, " +
                " `ts` bigint, " +
                " `data` map<string, string>, " +
                " `old` map<string, string>, " +
                " `pt` as proctime() " +
                ")" + SQLUtil.getKafkaSource(Constant.TOPIC_ODS_DB, groupId));
    }

}





